package net.bwie.douyin.dwd.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

public class KafkaEtlJob {

    public static void main(String[] args) {

        //1.表执行环境
        TableEnvironment tEnv = getTableEnv();

        //2.输入表
        creatInputTable(tEnv);


        //3-数据处理-select
        handle(tEnv);




    }


    /**
     * //3-数据处理-select
     * @param tEnv
     */
    private static void handle(TableEnvironment tEnv) {

        tEnv.sqlQuery(
                ""
        );


    }

    /**
     * //2.输入表
     * @param tEnv
     */
    private static void creatInputTable(TableEnvironment tEnv) {


        tEnv.executeSql("CREATE TABLE dwd_traffic_live_events_source (\n" +
                "    // 事件核心信息\n" +
                "    event_detail ROW<\n" +
                "    watch_duration INT COMMENT '用户本次观看直播时长（秒）',\n" +
                "    comment_id STRING COMMENT '评论唯一标识（格式：前缀+日期+序列号）',\n" +
                "    event_type STRING COMMENT '事件类型（comment=评论事件，exit=离开等）',\n" +
                "    content STRING COMMENT '评论原文（已脱敏）',\n" +
                "    create_time BIGINT COMMENT '评论时间戳（毫秒级）',\n" +
                "    live_count INT COMMENT '该评论被点赞次数'\n" +
                "    >,\n" +
                "\n" +
                "    // 用户画像\n" +
                "    user_info ROW<\n" +
                "    user_id STRING COMMENT '用户唯一标识（脱敏处理）',\n" +
                "    device_id STRING COMMENT '设备播放标识',\n" +
                "    gender INT COMMENT '性别（0=未成年男,1=男,2=女）',\n" +
                "    age_range STRING COMMENT '年龄分段',\n" +
                "    user_status STRING COMMENT '用户价值分层（viewer=普通观众/payer=付费用户）',\n" +
                "\n" +
                "    watch_session ROW<\n" +
                "    is_WatchInfo BOOLEAN COMMENT '本场直播观看行为',\n" +
                "    last_enter_time BIGINT COMMENT '最后一次进入时间戳',\n" +
                "    accumulative_duration INT COMMENT '本场累计观看时长（秒）'\n" +
                "    >,\n" +
                "\n" +
                "    first_traffic_source ROW<\n" +
                "    channel_type STRING COMMENT '首次触达来源',\n" +
                "    campaign_id STRING COMMENT '一级流量类型'\n" +
                "    >,\n" +
                "\n" +
                "    fans_culp_info ROW<\n" +
                "    is_member BOOLEAN COMMENT '是否粉丝团成员',\n" +
                "    is_follow BOOLEAN COMMENT '是否关注主播',\n" +
                "    join_time BIGINT COMMENT '加入粉丝团时间（秒级）'\n" +
                "    >\n" +
                "    >,\n" +
                "\n" +
                "    -- 商品数据\n" +
                "    product_data ROW<\n" +
                "    product_id STRING COMMENT '商品SPU编码',\n" +
                "    product_name STRING COMMENT '商品名称（含货号）',\n" +
                "    price INT COMMENT '单价（单位：分）',\n" +
                "    quantity INT COMMENT '购买数量',\n" +
                "    start_time BIGINT COMMENT '商品上架时间戳',\n" +
                "    end_time BIGINT COMMENT '商品下架时间戳',\n" +
                "    order_province_code STRING COMMENT '收货地省份代码'\n" +
                "    >,\n" +
                "\n" +
                "    -- 订单分析\n" +
                "    order_channel ROW<\n" +
                "    order_id STRING COMMENT '订单唯一标识',\n" +
                "    channel_type STRING COMMENT '订单来源类型',\n" +
                "    video_id STRING COMMENT '引流视频ID',\n" +
                "    ad_plan_id STRING COMMENT '广告计划ID',\n" +
                "    is_payment BOOLEAN COMMENT '支付状态'\n" +
                "    >,\n" +
                "\n" +
                "    -- 处理时间\n" +
                "    ts BIGINT COMMENT '数据处理时间戳（服务端时间）',\n" +
                "\n" +
                "    -- 计算列与水印\n" +
                "    -- 使用 COALESCE(event_detail.create_time, ts) 确保总能获取有效时间戳\n" +
                "    -- 事件时间(event_time)优先使用评论创建时间，不存在时使用服务端时间\n" +
                "    event_time AS TO_TIMESTAMP(FROM_UNIXTIME(COALESCE(event_detail.create_time, ts) / 1000)),\n" +
                "    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'douyin-topic',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "  'properties.group.id' = 'flink_consumer_group',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json',\n" +
                "  'json.ignore-parse-errors' = 'true',\n" +
                ");");

        // 2. 查询数据（不需要创建临时视图）
        TableResult result = tEnv.executeSql("SELECT * FROM dwd_traffic_live_events_source LIMIT 10");
        result.print();

    }


    /**
     * //1.表执行环境
     * @return
     */
    private static TableEnvironment getTableEnv() {

        //1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();

        TableEnvironment tEnv = TableEnvironment.create(settings);

        //2.配置属性设置
        Configuration configuration = tEnv.getConfig().getConfiguration();

        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");


        return tEnv;
    }
}
